package com.tool.engine.export.write.allocator;

import com.tool.engine.export.mertics.Collector;
import com.tool.engine.export.write.WriteTask;
import lombok.Setter;
import org.springframework.beans.factory.InitializingBean;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.function.UnaryOperator;
import java.util.zip.CRC32;

public class DataWriteAllocator implements InitializingBean {

    @Setter
    private int writeCount = 2;

    private List<DataWriteThread> writeThreads;

    @Setter
    private Collector collector;

    public void dispatcher(String fileName, WriteTask writeTask, UnaryOperator<Boolean> consumer) {
        writeThreads.get(getPartition(fileName, writeCount)).put(writeTask, consumer);
    }

    public int getPartition(String filename, int partitions) {
        CRC32 crc32 = new CRC32();
        crc32.update(filename.getBytes(StandardCharsets.UTF_8));
        return (int) (crc32.getValue() % partitions);
    }

    @Override
    public void afterPropertiesSet() {
        writeThreads = new ArrayList<>(writeCount);
        for (int i = 0; i < writeCount; i++) {
            DataWriteThread dataWriteThread = new DataWriteThread("dataWriteThread-" + i);
            dataWriteThread.setCollector(collector);
            dataWriteThread.start();
            writeThreads.add(dataWriteThread);
        }
    }
}
